package org.quxiao.mmrsql.executor.mapper;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.quxiao.mmrsql.executor.reducer.TableValueReducer;

public class TableValueMapper extends TableMapper<Text, Text>{

	private String tableName = "";
	private String familyName = "";
	private String qualifiedName = "";
	
	
	@Override
	protected void map(ImmutableBytesWritable key, Result value, Context context)
			throws IOException, InterruptedException {
		/*
		 * input key	:	id
		 * input value 	:	attribute
		 * output key	:	tableName:id
		 * output value	:	attribute
		 */
		String tableName = context.getConfiguration().get("tableName");
		String outKey = tableName + ':' + new String(key.get(), "utf8");
		String outValue = new String(value.value());
//		System.out.println(outKey + ' ' + outValue);
		context.write(new Text(outKey), new Text(outValue));
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception{
		// TODO Auto-generated method stub
		String tableName = "Asset_100000";
		String familyName = "MetaData";
		String qualifiedName = "vendor";
		
		Configuration conf = new Configuration();
		conf.set("tableName", tableName);
		Job job = new Job(conf, "TestTableInput");
		Scan scan = new Scan();
		scan.addColumn(familyName.getBytes(), qualifiedName.getBytes());
		TableMapReduceUtil.initTableMapperJob(tableName, scan, TableValueMapper.class, Text.class, Text.class, job);	
		job.setJarByClass(TableValueMapper.class);
		job.setReducerClass(TableValueReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		TextOutputFormat.setOutputPath(job, new Path("table_value_job_test_"+System.currentTimeMillis()));

		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}
